package com.sztx.pay.center.common.util;

import java.lang.reflect.Field;
import java.util.HashMap;
import java.util.Map;

import org.apache.hadoop.hbase.client.Put;

import com.sztx.pay.center.common.annotation.DataConverter;
import com.sztx.pay.center.common.annotation.HbaseColumn;
import com.sztx.pay.center.common.annotation.HbaseConverter;
import com.sztx.se.common.exception.BusinessException;
import com.sztx.se.dataaccess.hbase.serialize.HbaseSerializer;

public class HbaseUtils {

	private static Map<Class<? extends DataConverter>, DataConverter> CONVERTERS = new HashMap<Class<? extends DataConverter>, DataConverter>();

	public static Put buildPut(Object o, byte[] rowKey) throws IllegalAccessException {
		Put put = new Put(rowKey);
		Class<?> clazz = o.getClass();
		for (; clazz != Object.class; clazz = clazz.getSuperclass()) {
			Field[] fields = clazz.getDeclaredFields();
			Field.setAccessible(fields, true);
			for (Field field : fields) {
				field.setAccessible(true);
				if (field.isAnnotationPresent(HbaseColumn.class)) {
					HbaseColumn column = field.getAnnotation(HbaseColumn.class);
					String columnName = column.columnName();
					String familyName = column.familyName();
					if (field.get(o) == null) {
						continue;
					}
					byte[] value = convertObjectToBytes(field, o);
					if (value != null) {
						put.add(HbaseSerializer.serialize(familyName), HbaseSerializer.serialize(columnName), value);
					}
				}
			}
		}
		return put;
	}

	public static byte[] convertObjectToBytes(Field field, Object o) throws IllegalAccessException {
		byte[] value = null;
		if (field.isAnnotationPresent(HbaseConverter.class)) {
			return getConverter(field).from(o);
		}
		Class<?> type = field.getType();
		if (type == String.class) {
			value = HbaseSerializer.serialize(field.get(o).toString());
		} else if (type == Double.class || type == double.class) {
			value =HbaseSerializer.serialize((Double)field.get(o));
		} else if (type == Long.class || type == long.class) {
			value = HbaseSerializer.serialize((Long)field.get(o));
		} else if (type == Short.class || type == short.class) {
			value = HbaseSerializer.serialize((Short)field.get(o));
		} else if (type == Character.class || type == char.class) {
			value = HbaseSerializer.serialize((Character)field.get(o));
		} else if (type == Integer.class || type == int.class) {
			value = HbaseSerializer.serialize((Integer)field.get(o));
		} else if (type == Float.class || type == float.class) {
			value = HbaseSerializer.serialize((Float)field.get(o));
		} else if (type == Boolean.class || type == boolean.class) {
			value = HbaseSerializer.serialize((Boolean)field.get(o));
		} else if (type == Byte.class || type == byte.class) {
			value = HbaseSerializer.serialize((Byte)field.get(o));
		} else {
			value = HbaseSerializer.serialize(field.get(o).toString());
		}
		return value;
	}

	public static Object convertBytesToObject(Field field, byte[] bytes) throws IllegalAccessException {
        Object value = null;
        if (field.isAnnotationPresent(HbaseConverter.class)) {
            return getConverter(field).to(bytes);
        }
        Class<?> type = field.getType();
        value = HbaseSerializer.deserialize(type, bytes);
//        if (type == String.class) {
//            value = Bytes.toString(bytes);
//        } else if (type == Double.class || type == double.class) {
//            value = Bytes.toDouble(bytes);
//        } else if (type == Long.class || type == long.class) {
//            value = Bytes.toLong(bytes);
//        } else if (type == Short.class || type == short.class) {
//            value = Bytes.toShort(bytes);
//        } else if (type == Integer.class || type == int.class) {
//            value = Bytes.toInt(bytes);
//        } else if (type == Float.class || type == float.class) {
//            value = Bytes.toFloat(bytes);
//        } else if (type == Boolean.class || type == boolean.class) {
//            value = Bytes.toBoolean(bytes);
//        } else {
//            value = Bytes.toString(bytes);
//        }
        return value;
    }
	
	private static DataConverter getConverter(Field field) throws IllegalAccessException {
		HbaseConverter converter = field.getAnnotation(HbaseConverter.class);
		Class<? extends DataConverter> clazz = converter.value();
		DataConverter dataConverter = CONVERTERS.get(clazz);
		if (dataConverter == null) {
			try {
				dataConverter = clazz.newInstance();
				CONVERTERS.put(clazz, dataConverter);
			} catch (InstantiationException e) {
				throw new BusinessException("could't get instance of hbasconverter");
			}
		}
		return dataConverter;
	}

}
